package com.edee.bigday.order.config;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;

@Configuration
public class MyRabbitConfig {

    @Autowired
    RabbitTemplate rabbitTemplate;

    /**
     * 使用JSON序列化机制，进行消息转换
     * @return
     */
    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }

    /**
     * 定制RabbitTemplate
     *  1、服务器代理Broker收到消息，回调：
     *      1）、在配置文件中配置：spring.rabbitmq.publisher-confirm-type=correlated
     *      2）、设置确认回调：rabbitTemplate.setConfirmCallback
     *  2、消息未正确抵达队列，回调：
     *      1）#开启发送端消息抵达队列的确认机制：
     *      spring.rabbitmq.publisher-returns=true
     *      2）#开启下面的mandatory配置，作用是：只要抵达队列，就以异步方式优先回调上述returns确认：
     *      spring.rabbitmq.template.mandatory=true
     *      3）设置投递失败回调：rabbitTemplate.setReturnCallback
     *  3、接收端（消费端）确认
     *      保证消息被正确消费，然后才能从Broker的队列中删除消息。
     *      1）默认是自动确认的，只要消息被正确接收，客户端会自动确认，然后，服务端才会删除该消息。
     *          问题：队列收到了多条消息，自动Ack给服务器，但只有一个消息处理成功，然后发生了宕机，此时就会发生消息丢失。
     *          解决方案：手动确认。
     *          在配置文件中配置：
     *          spring.rabbitmq.listener.simple.acknowledge-mode=manual
     *          消费者手动确认模式下，只要没有明确告知MQ消息被可靠投递（Ack），则消息状态就一直是Unacked状态。
     *          此时，即使consumer宕机，消息也不会丢失，状态会变为Ready，等待有新的consumer客户端连接上之后，
     *          重新将消息发送给新客户端。
     *      2）如何签收消息？
     *          签收：
     *          使用Channel的basicAck()方法。
     *          void basicAck(long deliveryTag, boolean multiple) throws IOException;
     *          拒签：
     *              //requeue=false:将消息丢弃，requeue=true:将消息发回服务器，消息重新入队：
     *              //channel.basicNack(deliveryTag,false,true);
     *              channel.basicNack(deliveryTag,false,false);
     *              //void basicReject(long deliveryTag, boolean requeue) throws IOException;
     *              //channel.basicReject(deliveryTag,false);
     */
    @PostConstruct  //MyRabbitConfig对象创建完成，执行该方法
    public void initRabbitTemplate(){
        //设置确认回调：
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            /**
             * 1、只要消息抵达服务器代理Broker，ack就是true
             * @param correlationData  当前消息的唯一关联数据（消息的唯一Id）
             * @param ack  消息是否被成功消费
             * @param cause  失败原因
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                /**
                 * 299避免消息丢失，需要做到如下3点：
                 *  1、做好消息确认机制（publisher和consumer两端都要做好，consumer端要手动ACK）；
                 *  2、发送的每条消息都要在数据库中做好记录，定期将失败的消息进行重新发送。
                 */
                //服务器收到消息：
                System.out.println("confirm...correlationData:["+correlationData+"]===>ack:["+ack+"]" +
                        "===>cause:["+cause+"]");
            }
        });
        //设置消息抵达队列失败时的确认回调：
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            /**
             * 当消息未投递到指定队列时，才会触发该失败回调
             * @param message 投递失败的消息的详细信息
             * @param replyCode 回复的状态码
             * @param replyText 回复的文本内容
             * @param exchange 消息发往的交换器
             * @param routingKey 消息指定的路由键
             */
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                //299避免消息丢失：出现错误，修改数据库中消息的状态--->错误抵达
                System.out.println("失败的消息：【"+message+"】==>replyCode："+replyCode+"==>Exchange："+exchange+"==>routingKey:"+routingKey);
            }
        });
    }
}
